package com.hudson.namesrv.handler;


import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.domain.meta.BrokerInfo;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import com.hudson.namesrv.util.NettyUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;

public class BrokerMessageHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    public static final ConcurrentHashMap<String, String> BROKER_ADDRESS = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, Channel> BROKER_CHANNEL = new ConcurrentHashMap<>();
    private static final Logger logger = LoggerFactory.getLogger(BrokerMessageHandler.class);

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.debug("Broker 连接 NameSrv 成功");
        BROKER_CHANNEL.put(ctx.channel().id().asLongText(), ctx.channel());
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
        logger.debug("NameSrv 收到消息:" + msg);
        if (support(msg.getType())) {
            BrokerInfo brokerInfo = JSONUtil.fromJson(new String(msg.getBody(),StandardCharsets.UTF_8), BrokerInfo.class);
            BROKER_ADDRESS.put(brokerInfo.getBrokerName(), brokerInfo.getAddress() + ":" + brokerInfo.getPort());
            logger.debug("{} 成功传输信息到 NameSrv", brokerInfo.getBrokerName());
            NettyUtil.write(ctx.channel(), RemotingCommand.builder()
                    .type(MessageTypeEnum.SUCCESS_MESSAGE.getType())
                    .build());
            logger.debug("Broker 正常發送地址到 NameSrv");
        }

        ctx.fireChannelRead(msg);
    }

    public boolean support(String type) {
        return MessageTypeEnum.BROKER_INFO_MESSAGE.getType().equals(type);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.debug("Broker 连接 NameSrv 失败");
        BROKER_CHANNEL.remove(ctx.channel().id().asLongText());
    }
}
